//! # 消息持久化存储
//!
//! 提供消息的持久化存储、恢复和事务支持

use async_trait::async_trait;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use uuid::Uuid;

use crate::error::{MessageBusError, MessageBusResult};
use crate::message::Message;

/// 消息存储条目
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct MessageRecord {
    /// 消息ID
    pub id: String,
    /// 消息内容
    pub message: Message,
    /// 存储时间戳
    pub stored_at: chrono::DateTime<chrono::Utc>,
    /// 消息状态
    pub status: MessageStatus,
    /// 重试次数
    pub retry_count: u32,
    /// 下次重试时间
    pub next_retry_at: Option<chrono::DateTime<chrono::Utc>>,
    /// 过期时间
    pub expires_at: Option<chrono::DateTime<chrono::Utc>>,
    /// 元数据
    pub metadata: HashMap<String, String>,
}

/// 消息状态
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub enum MessageStatus {
    /// 待处理
    Pending,
    /// 处理中
    Processing,
    /// 已完成
    Completed,
    /// 失败
    Failed,
    /// 重试中
    Retrying,
    /// 已过期
    Expired,
    /// 已取消
    Cancelled,
}

/// 存储事务记录
#[derive(Debug)]
pub struct TransactionRecord {
    /// 事务ID
    pub id: String,
    /// 创建时间
    pub created_at: chrono::DateTime<chrono::Utc>,
    /// 事务状态
    pub status: TransactionStatus,
    /// 事务操作
    pub operations: Vec<TransactionOperation>,
}

/// 事务状态
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TransactionStatus {
    /// 活跃
    Active,
    /// 已提交
    Committed,
    /// 已回滚
    RolledBack,
    /// 超时
    Timeout,
}

/// 事务操作
#[derive(Debug, Clone)]
pub enum TransactionOperation {
    /// 存储消息
    Store(MessageRecord),
    /// 更新消息状态
    UpdateStatus(String, MessageStatus),
    /// 删除消息
    Delete(String),
    /// 批量操作
    Batch(Vec<TransactionOperation>),
}

/// 查询条件
#[derive(Debug, Clone, Default)]
pub struct QueryCriteria {
    /// 消息状态过滤
    pub status: Option<MessageStatus>,
    /// 主题过滤
    pub topic_pattern: Option<String>,
    /// 时间范围过滤
    pub time_range: Option<TimeRange>,
    /// 最大返回数量
    pub limit: Option<usize>,
    /// 偏移量
    pub offset: Option<usize>,
    /// 排序方式
    pub order_by: Option<OrderBy>,
}

/// 时间范围
#[derive(Debug, Clone)]
pub struct TimeRange {
    /// 开始时间
    pub start: chrono::DateTime<chrono::Utc>,
    /// 结束时间
    pub end: chrono::DateTime<chrono::Utc>,
}

/// 排序方式
#[derive(Debug, Clone)]
pub enum OrderBy {
    /// 按存储时间排序
    StoredAt(SortOrder),
    /// 按消息ID排序
    MessageId(SortOrder),
    /// 按重试次数排序
    RetryCount(SortOrder),
}

/// 排序顺序
#[derive(Debug, Clone)]
pub enum SortOrder {
    /// 升序
    Ascending,
    /// 降序
    Descending,
}

/// 存储统计信息
#[derive(Debug, Clone, Default)]
pub struct StorageStats {
    /// 总消息数
    pub total_messages: u64,
    /// 待处理消息数
    pub pending_messages: u64,
    /// 处理中消息数
    pub processing_messages: u64,
    /// 已完成消息数
    pub completed_messages: u64,
    /// 失败消息数
    pub failed_messages: u64,
    /// 存储大小（字节）
    pub storage_size: u64,
    /// 最旧消息时间
    pub oldest_message: Option<chrono::DateTime<chrono::Utc>>,
    /// 最新消息时间
    pub newest_message: Option<chrono::DateTime<chrono::Utc>>,
}

/// 消息存储接口
#[async_trait]
pub trait MessageStorage: Send + Sync {
    /// 存储消息
    async fn store(&self, message: Message) -> MessageBusResult<String>;

    /// 批量存储消息
    async fn store_batch(&self, messages: Vec<Message>) -> MessageBusResult<Vec<String>>;

    /// 获取消息
    async fn get(&self, message_id: &str) -> MessageBusResult<Option<MessageRecord>>;

    /// 查询消息
    async fn query(&self, criteria: QueryCriteria) -> MessageBusResult<Vec<MessageRecord>>;

    /// 更新消息状态
    async fn update_status(&self, message_id: &str, status: MessageStatus) -> MessageBusResult<()>;

    /// 删除消息
    async fn delete(&self, message_id: &str) -> MessageBusResult<()>;

    /// 批量删除消息
    async fn delete_batch(&self, message_ids: Vec<String>) -> MessageBusResult<()>;

    /// 清理过期消息
    async fn cleanup_expired(&self) -> MessageBusResult<u64>;

    /// 获取存储统计信息
    async fn get_stats(&self) -> MessageBusResult<StorageStats>;

    /// 开始事务
    async fn begin_transaction(&self) -> MessageBusResult<Box<dyn StorageTransaction>>;

    /// 健康检查
    async fn health_check(&self) -> MessageBusResult<()>;
}

/// 存储事务接口
#[async_trait]
pub trait StorageTransaction: Send + Sync {
    /// 获取事务ID
    fn id(&self) -> &str;

    /// 在事务中存储消息
    async fn store(&mut self, message: Message) -> MessageBusResult<String>;

    /// 在事务中更新消息状态
    async fn update_status(&mut self, message_id: &str, status: MessageStatus) -> MessageBusResult<()>;

    /// 在事务中删除消息
    async fn delete(&mut self, message_id: &str) -> MessageBusResult<()>;

    /// 提交事务
    async fn commit(self: Box<Self>) -> MessageBusResult<()>;

    /// 回滚事务
    async fn rollback(self: Box<Self>) -> MessageBusResult<()>;
}

/// 消息恢复器
pub struct MessageRecovery {
    storage: Arc<dyn MessageStorage>,
    recovery_config: RecoveryConfig,
}

/// 恢复配置
#[derive(Debug, Clone)]
pub struct RecoveryConfig {
    /// 是否启用自动恢复
    pub auto_recovery: bool,
    /// 恢复间隔（秒）
    pub recovery_interval_secs: u64,
    /// 批量恢复大小
    pub batch_size: usize,
    /// 最大重试次数
    pub max_retry_count: u32,
    /// 重试延迟策略
    pub retry_delay_strategy: RetryDelayStrategy,
}

/// 重试延迟策略
#[derive(Debug, Clone)]
pub enum RetryDelayStrategy {
    /// 固定延迟
    Fixed(std::time::Duration),
    /// 指数退避
    Exponential {
        base_delay: std::time::Duration,
        max_delay: std::time::Duration,
        multiplier: f64,
    },
    /// 线性增长
    Linear {
        base_delay: std::time::Duration,
        increment: std::time::Duration,
    },
}

impl MessageRecovery {
    /// 创建新的消息恢复器
    pub fn new(storage: Arc<dyn MessageStorage>, config: RecoveryConfig) -> Self {
        Self {
            storage,
            recovery_config: config,
        }
    }

    /// 启动恢复服务
    pub async fn start(&self) -> MessageBusResult<()> {
        if !self.recovery_config.auto_recovery {
            tracing::info!("自动恢复已禁用，跳过恢复服务启动");
            return Ok(());
        }

        let storage = self.storage.clone();
        let config = self.recovery_config.clone();

        tracing::info!(
            "正在启动消息恢复服务: 间隔={}s, 批大小={}, 最大重试次数={}",
            config.recovery_interval_secs,
            config.batch_size,
            config.max_retry_count
        );

        tokio::spawn(async move {
            let mut interval = tokio::time::interval(
                std::time::Duration::from_secs(config.recovery_interval_secs)
            );
            let mut cycle_count = 0u64;

            loop {
                interval.tick().await;
                cycle_count += 1;
                
                tracing::trace!("执行第 {} 次消息恢复周期", cycle_count);
                
                match Self::recovery_cycle(&storage, &config).await {
                    Ok(()) => {
                        tracing::trace!("第 {} 次恢复周期成功完成", cycle_count);
                    }
                    Err(e) => {
                        tracing::error!("第 {} 次恢复周期失败: {}", cycle_count, e);
                    }
                }
            }
        });

        tracing::info!("消息恢复服务已启动");
        Ok(())
    }

    /// 执行一次恢复周期
    async fn recovery_cycle(
        storage: &Arc<dyn MessageStorage>,
        config: &RecoveryConfig,
    ) -> MessageBusResult<()> {
        let start_time = std::time::Instant::now();
        
        // 查询需要恢复的消息
        let criteria = QueryCriteria {
            status: Some(MessageStatus::Failed),
            limit: Some(config.batch_size),
            order_by: Some(OrderBy::StoredAt(SortOrder::Ascending)),
            ..Default::default()
        };

        let failed_messages = storage.query(criteria).await?;
        let failed_count = failed_messages.len();
        
        tracing::debug!("恢复周期开始: 找到 {} 条失败消息需要处理", failed_count);
        
        let mut recovered_count = 0;
        let mut expired_count = 0;
        let mut skipped_count = 0;
        
        for record in failed_messages {
            if record.retry_count >= config.max_retry_count {
                // 超过最大重试次数，标记为过期
                tracing::warn!(
                    "消息超过最大重试次数，标记为过期: id={}, retry_count={}, max_retry={}",
                    record.id,
                    record.retry_count,
                    config.max_retry_count
                );
                storage.update_status(&record.id, MessageStatus::Expired).await?;
                expired_count += 1;
                continue;
            }

            // 检查是否到了重试时间
            if let Some(next_retry) = record.next_retry_at {
                if chrono::Utc::now() < next_retry {
                    tracing::trace!("消息未到重试时间，跳过: id={}", record.id);
                    skipped_count += 1;
                    continue;
                }
            }

            // 重新发布消息进行处理
            tracing::debug!("开始恢复消息: id={}, retry_count={}", record.id, record.retry_count);
            
            match storage.update_status(&record.id, MessageStatus::Retrying).await {
                Ok(()) => {
                    tracing::info!("消息恢复成功: id={}", record.id);
                    recovered_count += 1;
                }
                Err(e) => {
                    tracing::error!("消息恢复失败: id={}, error={}", record.id, e);
                }
            }
        }

        // 清理过期消息
        let cleaned_count = storage.cleanup_expired().await?;
        
        let duration = start_time.elapsed();
        
        if recovered_count > 0 || expired_count > 0 || cleaned_count > 0 {
            tracing::info!(
                "恢复周期完成: 恢复={}, 过期={}, 跳过={}, 清理={}, 耗时={:?}",
                recovered_count,
                expired_count,
                skipped_count,
                cleaned_count,
                duration
            );
        } else {
            tracing::trace!("恢复周期完成: 无需处理的消息, 耗时={:?}", duration);
        }

        Ok(())
    }

    /// 手动恢复指定消息
    pub async fn recover_message(&self, message_id: &str) -> MessageBusResult<()> {
        if let Some(record) = self.storage.get(message_id).await? {
            if matches!(record.status, MessageStatus::Failed | MessageStatus::Expired) {
                self.storage.update_status(message_id, MessageStatus::Pending).await?;
                tracing::info!("手动恢复消息: {}", message_id);
            }
        }
        Ok(())
    }

    /// 获取恢复统计信息
    pub async fn get_recovery_stats(&self) -> MessageBusResult<RecoveryStats> {
        let stats = self.storage.get_stats().await?;
        
        Ok(RecoveryStats {
            failed_messages: stats.failed_messages,
            retrying_messages: self.count_retrying_messages().await?,
            expired_messages: self.count_expired_messages().await?,
            recovery_enabled: self.recovery_config.auto_recovery,
        })
    }

    /// 统计重试中的消息数量
    async fn count_retrying_messages(&self) -> MessageBusResult<u64> {
        let criteria = QueryCriteria {
            status: Some(MessageStatus::Retrying),
            ..Default::default()
        };
        let messages = self.storage.query(criteria).await?;
        Ok(messages.len() as u64)
    }

    /// 统计过期消息数量
    async fn count_expired_messages(&self) -> MessageBusResult<u64> {
        let criteria = QueryCriteria {
            status: Some(MessageStatus::Expired),
            ..Default::default()
        };
        let messages = self.storage.query(criteria).await?;
        Ok(messages.len() as u64)
    }
}

/// 恢复统计信息
#[derive(Debug, Clone)]
pub struct RecoveryStats {
    /// 失败消息数
    pub failed_messages: u64,
    /// 重试中消息数
    pub retrying_messages: u64,
    /// 过期消息数
    pub expired_messages: u64,
    /// 是否启用恢复
    pub recovery_enabled: bool,
}

impl Default for RecoveryConfig {
    fn default() -> Self {
        Self {
            auto_recovery: true,
            recovery_interval_secs: 60,
            batch_size: 100,
            max_retry_count: 3,
            retry_delay_strategy: RetryDelayStrategy::Exponential {
                base_delay: std::time::Duration::from_secs(30),
                max_delay: std::time::Duration::from_secs(3600),
                multiplier: 2.0,
            },
        }
    }
}

impl MessageRecord {
    /// 创建新的消息记录
    pub fn new(message: Message) -> Self {
        Self {
            id: Uuid::new_v4().to_string(),
            message,
            stored_at: chrono::Utc::now(),
            status: MessageStatus::Pending,
            retry_count: 0,
            next_retry_at: None,
            expires_at: None,
            metadata: HashMap::new(),
        }
    }

    /// 设置过期时间
    pub fn with_expiry(mut self, expires_at: chrono::DateTime<chrono::Utc>) -> Self {
        self.expires_at = Some(expires_at);
        self
    }

    /// 添加元数据
    pub fn with_metadata(mut self, key: String, value: String) -> Self {
        self.metadata.insert(key, value);
        self
    }

    /// 检查是否过期
    pub fn is_expired(&self) -> bool {
        if let Some(expires_at) = self.expires_at {
            chrono::Utc::now() > expires_at
        } else {
            false
        }
    }

    /// 计算下次重试时间
    pub fn calculate_next_retry(&self, strategy: &RetryDelayStrategy) -> chrono::DateTime<chrono::Utc> {
        let delay = match strategy {
            RetryDelayStrategy::Fixed(duration) => *duration,
            RetryDelayStrategy::Exponential { base_delay, max_delay, multiplier } => {
                let delay_secs = base_delay.as_secs() as f64 * multiplier.powi(self.retry_count as i32);
                let delay = std::time::Duration::from_secs(delay_secs as u64);
                std::cmp::min(delay, *max_delay)
            }
            RetryDelayStrategy::Linear { base_delay, increment } => {
                *base_delay + *increment * self.retry_count
            }
        };

        chrono::Utc::now() + chrono::Duration::from_std(delay).unwrap_or(chrono::Duration::hours(1))
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn test_message_record_creation() {
        let message = Message::new("test.topic", json!({"test": "data"}));
        let record = MessageRecord::new(message.clone());

        assert_eq!(record.message.topic(), "test.topic");
        assert_eq!(record.status, MessageStatus::Pending);
        assert_eq!(record.retry_count, 0);
        assert!(!record.id.is_empty());
    }

    #[test]
    fn test_message_expiry() {
        let message = Message::new("test.topic", json!({"test": "data"}));
        let past_time = chrono::Utc::now() - chrono::Duration::hours(1);
        let future_time = chrono::Utc::now() + chrono::Duration::hours(1);

        let expired_record = MessageRecord::new(message.clone()).with_expiry(past_time);
        let valid_record = MessageRecord::new(message.clone()).with_expiry(future_time);

        assert!(expired_record.is_expired());
        assert!(!valid_record.is_expired());
    }

    #[test]
    fn test_retry_delay_calculation() {
        let message = Message::new("test.topic", json!({"test": "data"}));
        let mut record = MessageRecord::new(message);
        record.retry_count = 2;

        let fixed_strategy = RetryDelayStrategy::Fixed(std::time::Duration::from_secs(30));
        let next_retry = record.calculate_next_retry(&fixed_strategy);
        let expected = chrono::Utc::now() + chrono::Duration::seconds(30);
        
        // 允许1秒的误差
        assert!((next_retry - expected).num_seconds().abs() <= 1);

        let exp_strategy = RetryDelayStrategy::Exponential {
            base_delay: std::time::Duration::from_secs(10),
            max_delay: std::time::Duration::from_secs(300),
            multiplier: 2.0,
        };
        let next_retry_exp = record.calculate_next_retry(&exp_strategy);
        let expected_exp = chrono::Utc::now() + chrono::Duration::seconds(40); // 10 * 2^2 = 40
        
        assert!((next_retry_exp - expected_exp).num_seconds().abs() <= 1);
    }

    #[test]
    fn test_query_criteria() {
        let criteria = QueryCriteria {
            status: Some(MessageStatus::Failed),
            topic_pattern: Some("user.events.*".to_string()),
            limit: Some(100),
            ..Default::default()
        };

        assert_eq!(criteria.status, Some(MessageStatus::Failed));
        assert_eq!(criteria.topic_pattern, Some("user.events.*".to_string()));
        assert_eq!(criteria.limit, Some(100));
    }

    #[test]
    fn test_recovery_config_default() {
        let config = RecoveryConfig::default();
        
        assert!(config.auto_recovery);
        assert_eq!(config.recovery_interval_secs, 60);
        assert_eq!(config.batch_size, 100);
        assert_eq!(config.max_retry_count, 3);
        
        if let RetryDelayStrategy::Exponential { base_delay, max_delay, multiplier } = config.retry_delay_strategy {
            assert_eq!(base_delay, std::time::Duration::from_secs(30));
            assert_eq!(max_delay, std::time::Duration::from_secs(3600));
            assert_eq!(multiplier, 2.0);
        } else {
            panic!("Expected exponential retry delay strategy");
        }
    }
}